<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>第八章：Callbacks模块</title>
    <link rel="stylesheet" href="../shared_styles.css">
</head>
<body>
    <div class="container">
        <h1> 第八章：Callbacks模块</h1>

        <section id="s1">
            <h2>1. Callbacks模块简介</h2>
            <p>Langchain的Callbacks模块提供了一种强大的机制，允许开发者在LLM应用（如Chains、Agents、Models等）的生命周期的不同阶段插入自定义逻辑。这些回调函数可以用于日志记录、监控、流式处理、UI更新、数据收集等多种目的。</p>
            <p>当你运行一个Chain或Agent时，背后会发生很多事件，例如：</p>
            <ul>
                <li>Chain开始/结束执行</li>
                <li>LLM开始/结束调用</li>
                <li>工具开始/结束执行</li>
                <li>Agent做出决策（Action）</li>
                <li>检索器开始/结束检索</li>
                <li>产生错误</li>
            </ul>
            <p>Callbacks系统允许你“挂钩”到这些事件，并在它们发生时执行你自己的代码，而无需修改Langchain的内部实现。</p>
        </section>

        <section id="s2">
            <h2>2. Callback Handler (回调处理器)</h2>
            <p>核心概念是<strong>Callback Handler (回调处理器)</strong>。它是一个类，实现了一个或多个特定的回调方法，每个方法对应LLM应用生命周期中的一个事件。</p>
            <p>例如，一个回调处理器可能实现以下方法：</p>
            <ul>
                <li><code>on_llm_start(serialized, prompts, **kwargs)</code>: LLM调用开始时触发。</li>
                <li><code>on_llm_new_token(token, **kwargs)</code>: LLM流式输出新Token时触发。</li>
                <li><code>on_llm_end(response, **kwargs)</code>: LLM调用结束时触发。</li>
                <li><code>on_llm_error(error, **kwargs)</code>: LLM调用出错时触发。</li>
                <li><code>on_chain_start(serialized, inputs, **kwargs)</code>: Chain开始执行时触发。</li>
                <li><code>on_chain_end(outputs, **kwargs)</code>: Chain执行结束时触发。</li>
                <li><code>on_tool_start(serialized, input_str, **kwargs)</code>: 工具开始执行时触发。</li>
                <li><code>on_agent_action(action, **kwargs)</code>: Agent决定执行一个动作时触发。</li>
                <li>等等...</li>
            </ul>
            <p>你可以创建一个自定义的回调处理器，只需继承<code>langchain_core.callbacks.base.BaseCallbackHandler</code>并覆盖你感兴趣的事件方法。</p>
        </section>

        <section id="s3">
            <h2>3. 内置的回调处理器</h2>
            <p>Langchain提供了一些内置的回调处理器，方便快速使用：</p>
            <ul>
                <li><strong><code>StdOutCallbackHandler</code>:</strong> 将所有事件的日志信息打印到标准输出（控制台）。这对于调试非常有用。</li>
                <li><strong><code>FileCallbackHandler</code>:</strong> 将日志信息写入到指定的文件。</li>
                <!-- LangsmithCallbackHandler is for LangSmith, which is a separate service.
                     For general purpose, StdOut or custom handlers are more direct examples.
                <li><strong><code>LangChainTracer</code> (LangSmith):</strong> 与LangSmith平台集成，用于跟踪和可视化LLM应用的执行过程。</li>
                -->
            </ul>
            <p>我们主要关注<code>StdOutCallbackHandler</code>和如何创建自定义处理器。</p>
            <p><strong>示例：使用StdOutCallbackHandler</strong></p>
            <pre><code class="language-python">
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain.chains import LLMChain # 旧的LLMChain，用于演示callbacks参数，LCEL中callbacks通过config传递
from langchain_core.callbacks.stdout import StdOutCallbackHandler # 导入StdOutCallbackHandler
from dotenv import load_dotenv
import os

# 加载环境变量并初始化Qwen模型
load_dotenv()
api_key = os.environ.get("DASHSCOPE_API_KEY")
if not api_key:
    raise ValueError("DASHSCOPE_API_KEY not found in environment variables.")

qwen_llm = ChatOpenAI(
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    api_key=api_key,
    model="qwen-plus", 
    temperature=0.7,
    # streaming=True # 对于某些回调如on_llm_new_token需要开启流式输出
)

# 创建一个回调处理器实例
stdout_handler = StdOutCallbackHandler()

# ---- 使用旧式LLMChain演示callbacks参数 ----
# 提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个乐于助人的AI。"),
    ("human", "请问地球的周长大约是多少公里？")
])

# 创建LLMChain时传入callbacks列表 (旧方法)
# chain_with_stdout_callback_old = LLMChain(
#     llm=qwen_llm, 
#     prompt=prompt, 
#     callbacks=[stdout_handler], # 直接传递回调实例
#     verbose=False # verbose=True也会打印很多信息，可能会与stdout_handler重复
# )
# print("--- 使用StdOutCallbackHandler (通过LLMChain的callbacks参数) ---")
# response_old = chain_with_stdout_callback_old.invoke({}) # LLMChain 的 invoke 通常需要输入变量字典
# print(f"\nAI的最终回答: {response_old['text']}")


# ---- 使用LCEL演示通过 .invoke() 或 .stream() 的 config 传递callbacks ----
# LCEL 链
lcel_chain = prompt | qwen_llm

print("\n--- 使用StdOutCallbackHandler (通过LCEL的config参数) ---")
# 在调用invoke或stream时，通过config参数传递回调
# 对于同步调用 invoke
print("\n--- 同步调用 (invoke) ---")
response_lcel = lcel_chain.invoke(
    input={}, # LCEL链的输入，如果prompt没有input_variables则为空字典
    config={"callbacks": [stdout_handler]} # 通过config传递回调
)
print(f"\nAI的最终回答 (LCEL): {response_lcel.content}")


# 对于流式调用 stream (需要LLM初始化时 streaming=True)
# qwen_llm_streaming = ChatOpenAI(
#     base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
#     api_key=api_key,
#     model="qwen-plus", 
#     temperature=0.7,
#     streaming=True # 必须开启流式模式才能看到 on_llm_new_token
# )
# lcel_chain_streaming = prompt | qwen_llm_streaming
# print("\n--- 流式调用 (stream) ---")
# final_streamed_response = ""
# for chunk in lcel_chain_streaming.stream({}, config={"callbacks": [stdout_handler]}):
#     final_streamed_response += chunk.content
#     # print(chunk.content, end="", flush=True) # StdOutCallbackHandler会处理打印
# print(f"\nAI的最终流式回答 (LCEL): {final_streamed_response}")
            </code></pre>
            <p>当运行这段代码时，<code>StdOutCallbackHandler</code>会将Chain执行的各个阶段信息（如LLM调用开始、LLM调用结束、Token流等）打印到控制台。</p>
        </section>

        <section id="s4">
            <h2>4. 自定义回调处理器</h2>
            <p>创建自定义回调处理器非常简单，只需继承<code>BaseCallbackHandler</code>并实现你感兴趣的事件方法。</p>
            <p><strong>示例：一个记录LLM调用次数和Token使用量的自定义回调处理器</strong></p>
            <pre><code class="language-python">
from langchain_core.callbacks.base import BaseCallbackHandler
from langchain_core.outputs import LLMResult
from typing import Any, Dict, List, Optional, Union
from uuid import UUID

class MyCustomCallbackHandler(BaseCallbackHandler):
    def __init__(self):
        super().__init__()
        self.llm_calls = 0
        self.total_tokens_used = 0 # 粗略估计，需解析LLMResult
        self.llm_errors = 0
        self.last_prompt = ""
        self.last_response = ""

    def on_llm_start(
        self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any
    ) -> Any:
        """LLM开始运行时调用。"""
        self.llm_calls += 1
        self.last_prompt = prompts[0] # 假设只有一个prompt
        print(f"\n[MyCustomHandler] LLM Call #{self.llm_calls} - START")
        print(f"[MyCustomHandler] Prompt:\n{prompts[0][:200]}...") # 打印部分提示

    def on_llm_new_token(self, token: str, **kwargs: Any) -> Any:
        """当LLM流式输出新Token时调用。"""
        # print(f"[MyCustomHandler] New Token: {token}", end="", flush=True) # 实时打印Token
        pass # 通常StdOutCallbackHandler会做得更好

    def on_llm_end(self, response: LLMResult, **kwargs: Any) -> Any:
        """LLM成功结束运行时调用。"""
        self.last_response = response.generations[0][0].text # 假设只有一个生成结果
        print(f"\n[MyCustomHandler] LLM Call #{self.llm_calls} - END")
        print(f"[MyCustomHandler] Response:\n{self.last_response[:200]}...") # 打印部分响应
        
        # 尝试从LLMResult中获取Token使用信息
        if response.llm_output and 'token_usage' in response.llm_output:
            tokens = response.llm_output['token_usage'].get('total_tokens', 0)
            self.total_tokens_used += tokens
            print(f"[MyCustomHandler] Tokens used in this call: {tokens}")
        print(f"[MyCustomHandler] Total LLM calls so far: {self.llm_calls}")
        print(f"[MyCustomHandler] Total tokens used so far (approx): {self.total_tokens_used}")


    def on_llm_error(
        self, error: Union[Exception, KeyboardInterrupt], **kwargs: Any
    ) -> Any:
        """LLM运行出错时调用。"""
        self.llm_errors += 1
        print(f"\n[MyCustomHandler] LLM Error Occurred:")
        print(error)
        print(f"[MyCustomHandler] Total LLM errors: {self.llm_errors}")

    def on_chain_start(
        self, serialized: Dict[str, Any], inputs: Dict[str, Any], **kwargs: Any
    ) -> Any:
        """Chain开始运行时调用。"""
        print(f"\n[MyCustomHandler] Chain '{serialized.get('name', serialized.get('id', ['<unknown_chain>'])[0])}' - START")
        # print(f"[MyCustomHandler] Chain Inputs: {inputs}")

    def on_chain_end(self, outputs: Dict[str, Any], **kwargs: Any) -> Any:
        """Chain成功结束运行时调用。"""
        # print(f"\n[MyCustomHandler] Chain Outputs: {outputs}")
        print(f"[MyCustomHandler] Chain - END")
        
    def on_agent_action(self, action, **kwargs: Any) -> Any:
        """当Agent决定执行一个Action时调用。"""
        print(f"\n[MyCustomHandler] Agent Action: {action.tool} with input {action.tool_input}")
        # print(f"[MyCustomHandler] Agent log: {action.log}")


# 使用自定义回调处理器
my_handler = MyCustomCallbackHandler()

# (qwen_llm 和 prompt 已定义)
lcel_chain_custom_cb = prompt | qwen_llm # 使用之前的LCEL链

print("\n\n--- 使用MyCustomCallbackHandler (通过LCEL的config参数) ---")
custom_response = lcel_chain_custom_cb.invoke(
    {}, # 输入
    config={"callbacks": [my_handler, StdOutCallbackHandler()]} # 可以传递多个回调
) 
# print(f"\nAI的最终回答 (自定义回调): {custom_response.content}")

# 再次调用以观察计数器变化
print("\n\n--- 第二次调用 (观察计数器) ---")
custom_response_2 = lcel_chain_custom_cb.invoke(
    {"input": "光速是多少？"}, # 假设prompt现在接受input
    config={"callbacks": [my_handler]} # 这次只用自定义的
)
# print(f"\nAI的最终回答 (自定义回调2): {custom_response_2.content}")

print(f"\n--- 自定义处理器最终统计 ---")
print(f"总LLM调用次数: {my_handler.llm_calls}")
print(f"总Token估算: {my_handler.total_tokens_used}")
print(f"总LLM错误数: {my_handler.llm_errors}")

            </code></pre>
            <p class="note">请注意，<code>LLMResult.llm_output</code>中<code>token_usage</code>的具体结构可能因LLM提供商和模型而异。对于通过OpenAI兼容接口（如DashScope Qwen）调用的模型，通常会包含类似OpenAI的Token使用统计。对于其他模型，你可能需要检查其具体的<code>llm_output</code>结构。</p>
            <p>你可以根据需要在自定义回调处理器中实现更多的方法，以捕获更多类型的事件。</p>
        </section>

        <section id="s5">
            <h2>5. 回调的用途</h2>
            <p>Callbacks模块非常灵活，可以用于多种场景：</p>
            <ul>
                <li><strong>日志记录 (Logging):</strong> 记录Chain/Agent的详细执行步骤、输入输出、错误等，用于调试和审计。</li>
                <li><strong>监控 (Monitoring):</strong> 跟踪LLM的调用次数、Token使用量、响应时间、错误率等，以便监控应用性能和成本。</li>
                <li><strong>流式处理 (Streaming to UI):</strong> 通过<code>on_llm_new_token</code>回调，可以实时地将LLM生成的Token流式传输到用户界面，提升用户体验。</li>
                <li><strong>UI更新:</strong> 在Chain或Agent执行的不同阶段更新UI元素，例如显示加载状态、Agent的思考过程等。</li>
                <li><strong>数据收集/持久化:</strong> 将特定的输入、输出或中间结果保存到数据库或文件中，用于分析或后续处理。</li>
                <li><strong>调试:</strong> 打印详细的中间步骤和数据流，帮助理解和调试复杂的Chain或Agent。</li>
                <li><strong>与外部系统集成:</strong> 例如，在工具执行失败时发送通知，或将LLM的输出发送到另一个服务。</li>
                <li><strong>成本控制:</strong> 通过监控Token使用量，可以实现一些基本的成本预警或控制逻辑。</li>
            </ul>
        </section>

        <section id="s6">
            <h2>6. 如何传递和管理Callbacks</h2>
            <p>有多种方式可以将回调处理器关联到Langchain对象：</p>
            <ol>
                <li><strong>构造时传递 (Constructor):</strong> 一些旧的Langchain类（如某些Chain的实现）允许在构造函数中通过<code>callbacks</code>参数传递一个回调处理器列表。
                    <pre><code class="language-python">
# my_chain = SomeOldChainType(llm=my_llm, callbacks=[my_handler])
                    </code></pre>
                </li>
                <li><strong>通过<code>invoke</code>, <code>stream</code>, <code>batch</code>等的<code>config</code>参数 (推荐方式，尤其对于LCEL):</strong>
                    <p>对于使用Langchain表达式语言（LCEL）构建的链，以及许多现代的Langchain对象，推荐在调用<code>.invoke()</code>, <code>.stream()</code>, <code>.batch()</code>, <code>.ainvoke()</code>等方法时，通过<code>config</code>参数传入回调。</p>
                    <pre><code class="language-python">
# my_runnable = prompt | llm
# result = my_runnable.invoke(input_data, config={"callbacks": [handler1, handler2]})

# async_result = await my_runnable.ainvoke(input_data, config={"callbacks": [handler1]})

# for chunk in my_runnable.stream(input_data, config={"callbacks": [handler_for_streaming]}):
#     # process chunk
#     pass
                    </code></pre>
                    <p><code>config</code>是一个字典，<code>"callbacks"</code>键对应一个回调处理器列表。这种方式更加灵活，允许为单次调用指定特定的回调。</p>
                </li>
                <li><strong>全局回调 (Global Callbacks - 通常不推荐用于生产):</strong>
                    <p>可以通过<code>langchain_core.globals.set_llm_cache</code>和<code>langchain_core.globals.set_verbose</code>等函数设置一些全局行为，但直接设置全局回调处理器通常不直接提供，以避免副作用。推荐使用调用时的<code>config</code>参数。</p>
                </li>
                <li><strong>CallbackManager:</strong>
                    <p>在Langchain的内部，<code>CallbackManager</code>负责管理和分发事件给注册的回调处理器。通常开发者不需要直接与<code>CallbackManager</code>交互，除非在构建非常底层的自定义组件时。当你在<code>config</code>中传递回调时，Langchain内部会处理它们的注册和事件路由。</p>
                </li>
            </ol>
            <p>使用<code>config={"callbacks": ...}</code>的方式是最现代和推荐的做法，因为它提供了最大的灵活性和控制力，并且与LCEL无缝集成。</p>
        </section>
        
        <section id="s7">
            <h2>7. 总结与后续</h2>
            <p>Callbacks模块是Langchain中一个强大且底层的特性，它使得开发者能够深入了解和控制LLM应用的执行流程。通过实现自定义回调处理器，你可以轻松地为你的应用添加日志、监控、流式处理等高级功能。</p>
            <p>理解和善用Callbacks将帮助你构建更健壮、更易于调试和更具交互性的Langchain应用程序。</p>
            <p>在后续的实际应用案例中，我们可能会间接或直接地利用回调来实现特定的功能，例如在Agent执行时显示其思考过程。</p>
        </section>

        <div class="navigation">
            <a href="../langchain_tutorial_outline.html">返回教程目录</a>
            <a href="../chapter7/chapter7_indexes.html">上一章：Indexes模块</a>
            <!-- <a href="../chapter9/chapter9_applications.html">下一章：实际应用案例</a> -->
        </div>
    </div>
</body>
</html>
